#ifndef CONSUMER_H
#define CONSUMER_H
#include "../common/message.pb.h"
#include "../common/Log.hpp"
#include <unordered_map>
#include <mutex>
#include <memory>
#include <vector>
#include <functional>

using ConsumerCallback = function<void(const string, const MessageQueue::BasicProperties *bp, const string)>;

// 消费者信息结构
struct Consumer 
{
    using ptr = shared_ptr<Consumer>;
    Consumer()
    {
        lg(Debug, "创建新的消费者：%p", this);
    }

    Consumer(const string& ctag, const string& queue_name, bool ack_flag, const ConsumerCallback& cb)
        : tag(ctag), qname(queue_name), auto_ack(ack_flag), callback(std::move(cb))
    {
        lg(Debug, "创建新的消费者：%p", this);
    }

    ~Consumer()
    {
        lg(Debug, "移除消费者：%p", this);
    }

    string tag;
    string qname;
    bool auto_ack;
    ConsumerCallback callback;
};

// 以队列为单元进行消费者管理
class QueueConsumer
{
public:
    using ptr = shared_ptr<QueueConsumer>;
    QueueConsumer(const string& qname)
        : _qname(qname), _rr_seq(0)
    {}

    // 新增消费者
    Consumer::ptr Create(const string& ctag, const string& queue_name, bool ack_flag, const ConsumerCallback& cb)
    {
        // 1. 加锁
        unique_lock<mutex> lock(_mutex);
        // 2. 判断是否重复
        for(auto& consumer : _consumers)
        {
            if(consumer->tag == ctag)
                return Consumer::ptr();
        }
        // 3. 构造对象
        auto consumer = make_shared<Consumer>(ctag, queue_name, ack_flag, cb);
        // 4. 添加管理
        _consumers.push_back(consumer);
        return consumer;
    }

    // 从队列移除消费者
    void Remove(const string& ctag)
    {
        // 1. 加锁
        unique_lock<mutex> lock(_mutex);
        // 2. 遍历删除
        for(auto it = _consumers.begin(); it != _consumers.end(); it++)
        {
            if((*it)->tag == ctag)
            {
                _consumers.erase(it);
                return;
            }
        }
        return;
    }

    // 从队列中获取消费者
    Consumer::ptr Choose()
    {
        // 1. 加锁
        unique_lock<mutex> lock(_mutex);
        if(_consumers.empty())
            return Consumer::ptr();
        // 2. 获取轮转下标
        int idx = _rr_seq % _consumers.size();
        _rr_seq++;
        // 3. 获取对象返回
        return _consumers[idx];
    }

    // 判断是否为空
    bool Empty()
    {
        // 1. 加锁
        unique_lock<mutex> lock(_mutex);
        // 2. 获取
        return _consumers.empty();
    }

    // 判断指定消费者是否存在
    bool Exists(const string& ctag)
    {
        // 1. 加锁
        unique_lock<mutex> lock(_mutex);
        // 2. 遍历查找
        for(auto it = _consumers.begin(); it != _consumers.end(); it++)
            if((*it)->tag == ctag)
                return true;
        return false;
    }

    // 清理所有消费者
    void clear()
    {
        // 1. 加锁
        unique_lock<mutex> lock(_mutex);
        // 2. 清理
        _consumers.clear();
        _rr_seq = 0;
    }

private:
    string _qname;
    mutex _mutex;
    uint64_t _rr_seq;
    vector<Consumer::ptr> _consumers;
};

// 消费者统一管理结构
class ConsumerManager
{
public:
    using ptr = shared_ptr<ConsumerManager>;
    ConsumerManager()
    {}

    void InitQueueConsumer(const string& qname)
    {
        // 1. 加锁
        unique_lock<mutex> lock(_mutex);
        // 2. 重复判断
        auto it = _qconsumers.find(qname);
        if(it != _qconsumers.end())
            return;
        // 3. 新增
        auto qconsumers = make_shared<QueueConsumer>(qname);
        _qconsumers.insert(make_pair(qname, qconsumers));
    }

    void DestroyQueueConsumer(const string &qname) 
    {
        unique_lock<mutex> lock(_mutex);
        _qconsumers.erase(qname);
    }

    Consumer::ptr Create(const string &ctag, const string &queue_name,  bool ack_flag, const ConsumerCallback &cb) 
    {
        //获取队列的消费者管理单元句柄，通过句柄完成新建
        QueueConsumer::ptr qcp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end()) 
            {
                lg(LOG_ERROR_CODE, "没有找到队列 %s 的消费者管理句柄！", queue_name.c_str());
                return Consumer::ptr();
            }
            qcp = it->second;
        }
        return qcp->Create(ctag, queue_name, ack_flag, cb);
    }

    void remove(const string &ctag, const string &queue_name) 
    {
        QueueConsumer::ptr qcp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end()) {
                lg(LOG_ERROR_CODE, "没有找到队列 %s 的消费者管理句柄！", queue_name.c_str());
                return ;
            }
            qcp = it->second;
        }
        return qcp->Remove(ctag);
    }

    Consumer::ptr choose(const string &queue_name) 
    {
        QueueConsumer::ptr qcp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end()) {
                lg(LOG_ERROR_CODE, "没有找到队列 %s 的消费者管理句柄！", queue_name.c_str());
                return Consumer::ptr();
            }
            qcp = it->second;
        }
        return qcp->Choose();
    }

    bool empty(const string &queue_name) 
    {
        QueueConsumer::ptr qcp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end()) {
                lg(LOG_ERROR_CODE, "没有找到队列 %s 的消费者管理句柄！", queue_name.c_str());
                return false;
            }
            qcp = it->second;
        }
        return qcp->Empty();
    }

    bool exists(const string &ctag, const string &queue_name) 
    {
        QueueConsumer::ptr qcp;
        {
            unique_lock<mutex> lock(_mutex);
            auto it = _qconsumers.find(queue_name);
            if (it == _qconsumers.end()) {
                lg(LOG_ERROR_CODE, "没有找到队列 %s 的消费者管理句柄！", queue_name.c_str());
                return false;
            }
            qcp = it->second;
        }
        return qcp->Exists(ctag);
    }

    void clear() 
    {
        unique_lock<mutex> lock(_mutex);
        _qconsumers.clear();
    }
private:
    mutex _mutex;
    unordered_map<string, QueueConsumer::ptr> _qconsumers;
};

#endif